package com.dahua.messagesyncsh.controller;


import com.alibaba.fastjson2.JSON;
import com.dahua.messagesyncsh.pojo.CMode;
import com.dahua.messagesyncsh.pojo.Feature;
import com.dahua.messagesyncsh.pojo.SyncMessage;
import com.dahua.messagesyncsh.pojo.T485;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.*;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.util.StringUtils;
import org.springframework.web.bind.annotation.*;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.concurrent.ExecutionException;

@RestController
@RequestMapping("/kafka")
@Slf4j
public class KafkaController {
    @Resource
    private AdminClient adminClient;
    @Resource
    private KafkaTemplate<String, Object> kafkaTemplate;

    /*  外部同步测点信息接口*/
    @PostMapping("/cmode")
    public void cmode(@RequestBody CMode cmode) {
        kafkaTemplate.send("CMODE_DATA", JSON.toJSONString(cmode));
    }

    @PostMapping("/feature")
    public void feature(@RequestBody Feature feature) {
        kafkaTemplate.send("FEATURE_DATA", JSON.toJSONString(feature));
    }

    /*  外部同步报警接口*/
    @PostMapping("/sync")
    public void sync(@RequestBody SyncMessage syncMessage) {
        kafkaTemplate.send("SYNC_DATA", JSON.toJSONString(syncMessage));
    }

    @GetMapping("{topic}")
    public String message(@PathVariable("topic") String topic, @RequestBody T485 t485) {
        try{
            kafkaTemplate.send(topic, t485.toString());
        }catch (Exception e){
            return  JSON.toJSONString(e);
        }

        return "生产者生产消息：msg={" + JSON.toJSONString(t485) + "}";
    }

    @GetMapping("data/{topic}")
    public String toData(@PathVariable("topic") String topic, @RequestBody Object data) {
        try{
            ObjectMapper objectMapper = new ObjectMapper();
            switch (topic) {
                case "FEATURE_DATA":
                    Feature feature = objectMapper.convertValue(data, Feature.class);
                    kafkaTemplate.send("FEATURE_DATA", JSON.toJSONString(feature));
                    break;
                case "CMODE_DATA":
                    CMode cmode = objectMapper.convertValue(data, CMode.class);
                    kafkaTemplate.send("CMODE_DATA", JSON.toJSONString(cmode));
                    break;
                case "CONTROL_DATA":
                    T485 t485 = objectMapper.convertValue(data, T485.class);
                    kafkaTemplate.send("CONTROL_DATA", JSON.toJSONString(t485));
                    break;
            }
        }catch (Exception e){
            return  JSON.toJSONString(e);
        }

        return "生产者生产消息：msg={" + JSON.toJSONString(data) + "}";
    }

    @GetMapping("/topic/list")
    public String list() throws ExecutionException, InterruptedException {
        ListTopicsOptions listTopicsOptions = new ListTopicsOptions();
        listTopicsOptions.listInternal(true);
        ListTopicsResult result = adminClient.listTopics(listTopicsOptions);
        Collection<TopicListing> list = result.listings().get();
        return list.toString();
    }

    @PostMapping("/topic/create")
    public void create(@RequestBody String topics) {
        if (StringUtils.isEmpty(topics)) {
            return;
        }
        Collection<NewTopic> newTopics = new ArrayList<>();
        Arrays.stream(topics.split(",")).forEach(topic -> {
            NewTopic newTopic = new NewTopic(topic, 1, (short) 1);
            newTopics.add(newTopic);
        });
        adminClient.createTopics(newTopics);
    }
}
